package thinkingInJava.concurrency;

//: concurrency/E29_ToastOMatic2.java
/********************** Exercise 29 ***********************
 * Modify ToastOMatic.java to create peanut butter and jelly
 * on toast sandwiches using two separate assembly lines
 * (one for peanut butter, the second for jelly, then
 * merging the two lines).
 *********************************************************/
import java.util.concurrent.*;
import java.util.*;
import static thinkingInJava.YTool.YPrint.*;
class Toast2 {
	public enum Status {
		DRY,
		BUTTERED,
		JAMMED,
		READY {
			public String toString() {
				return
						BUTTERED.toString() + " & " + JAMMED.toString();
			}
		}
	}
	private Status status = Status.DRY;
	private final int id;
	public Toast2(int idn) { id = idn; }
	public void butter() {
		status =
				(status == Status.DRY) ? Status.BUTTERED :
					Status.READY;
	}
	public void jam() {
		status =
				(status == Status.DRY) ? Status.JAMMED :
					Status.READY;
	}
	public Status getStatus() { return status; }
	public int getId() { return id; }
	public String toString() {
		return "Toast2 " + id + ": " + status;
	}
}
class ToastQueue extends LinkedBlockingQueue<Toast2> {}
class Toaster2 implements Runnable {
	private ToastQueue toastQueue;
	private int count;
	private Random rand = new Random(47);
	public Toaster2(ToastQueue tq) { toastQueue = tq; }
	public void run() {
		try {
			while(!Thread.interrupted()) {
				TimeUnit.MILLISECONDS.sleep(
						100 + rand.nextInt(500));
				// Make toast
				Toast2 t = new Toast2(count++);
				print(t);
				// Insert into queue
				toastQueue.put(t);
			}
		} catch(InterruptedException e) {
			print("Toaster2 interrupted");
		}
		print("Toaster2 off");
	}
}
//Apply butter to toast:
class Butterer2 implements Runnable {
	private ToastQueue inQueue, butteredQueue;
	public Butterer2(ToastQueue in, ToastQueue buttered) {
		inQueue = in;
		butteredQueue = buttered;
	}
	public void run() {
		try {
			while(!Thread.interrupted()) {
				// Blocks until next piece of toast is available:
				Toast2 t = inQueue.take();
				t.butter();
				print(t);
				butteredQueue.put(t);
			}
		} catch(InterruptedException e) {
			print("Butterer2 interrupted");
		}
		print("Butterer2 off");
	}
}
//Apply jam to toast:
class Jammer2 implements Runnable {
	private ToastQueue inQueue, jammedQueue;
	public Jammer2(ToastQueue in, ToastQueue jammed) {
		inQueue = in;
		jammedQueue = jammed;
	}
	public void run() {
		try {
			while(!Thread.interrupted()) {
				// Blocks until next piece of toast is available:
				Toast2 t = inQueue.take();
				t.jam();
				print(t);
				jammedQueue.put(t);
			}
		} catch(InterruptedException e) {
			print("Jammer2 interrupted");
		}
		print("Jammer2 off");
	}
}
//Consume the toast:
class Eater2 implements Runnable {
	private ToastQueue finishedQueue;
	public Eater2(ToastQueue finished) {
		finishedQueue = finished;
	}
	public void run() {
		try {
			while(!Thread.interrupted()) {
				// Blocks until next piece of toast is available:
				Toast2 t = finishedQueue.take();
				// Verify that all pieces are ready for consumption:
				if(t.getStatus() != Toast2.Status.READY) {
					print(">>>> Error: " + t);
					System.exit(1);
				} else
					print("Chomp! " + t);
			}
		} catch(InterruptedException e) {
			print("Eater2 interrupted");
		}
		print("Eater2 off");
	}
}
//Outputs alternate inputs on alternate channels:
class Alternator implements Runnable {
	private ToastQueue inQueue, out1Queue, out2Queue;
	private boolean outTo2; // control alternation
	public Alternator(ToastQueue in, ToastQueue out1,
			ToastQueue out2) {
		inQueue = in;
		out1Queue = out1;
		out2Queue = out2;
	}
	public void run() {
		try {
			while(!Thread.interrupted()) {
				// Blocks until next piece of toast is available:
				Toast2 t = inQueue.take();
				if(!outTo2)
					out1Queue.put(t);
				else
					out2Queue.put(t);
				outTo2 = !outTo2; // change state for next time
			}
		} catch(InterruptedException e) {
			print("Alternator interrupted");
		}
		print("Alternator off");
	}
}
//Accepts toasts on either channel, and relays them on to
//a "single" successor
class Merger implements Runnable {
	private ToastQueue in1Queue, in2Queue, toBeButteredQueue,
	toBeJammedQueue, finishedQueue;
	public Merger(ToastQueue in1, ToastQueue in2,
			ToastQueue toBeButtered, ToastQueue toBeJammed,
			ToastQueue finished) {
		in1Queue = in1;
		in2Queue = in2;
		toBeButteredQueue = toBeButtered;
		toBeJammedQueue = toBeJammed;
		finishedQueue = finished;
	}
	public void run() {
		try {
			while(!Thread.interrupted()) {
				// Blocks until next piece of toast is available:
				Toast2 t = null;
				while(t == null) {
					t = in1Queue.poll(50, TimeUnit.MILLISECONDS);
					if(t != null)
						break;
					t = in2Queue.poll(50, TimeUnit.MILLISECONDS);
				}
				// Relay toast onto the proper queue
				switch(t.getStatus()) {
				case BUTTERED:
					toBeJammedQueue.put(t);
					break;
				case JAMMED:
					toBeButteredQueue.put(t);
					break;
				default:
					finishedQueue.put(t);
				}
			}
		} catch(InterruptedException e) {
			print("Merger interrupted");
		}
		print("Merger off");
	}
}
public class E29_ToastOMatic2 {
	public static void main(String[] args) throws Exception {
		ToastQueue
		dryQueue = new ToastQueue(),
		butteredQueue = new ToastQueue(),
		toBeButteredQueue = new ToastQueue(),
		jammedQueue = new ToastQueue(),
		toBeJammedQueue = new ToastQueue(),
		finishedQueue = new ToastQueue();
		ExecutorService exec = Executors.newCachedThreadPool();
		exec.execute(new Toaster2(dryQueue));
		exec.execute(new Alternator(dryQueue, toBeButteredQueue,
				toBeJammedQueue));
		exec.execute(
				new Butterer2(toBeButteredQueue, butteredQueue));
		exec.execute(
				new Jammer2(toBeJammedQueue, jammedQueue));
		exec.execute(new Merger(butteredQueue , jammedQueue,
				toBeButteredQueue, toBeJammedQueue, finishedQueue));
		exec.execute(new Eater2(finishedQueue));
		TimeUnit.SECONDS.sleep(5);
		exec.shutdownNow();
	}
} /* (Execute to see output) *///:~
//The finished sandwiches no longer come in predefined order because the two
//assembly lines work in parallel. Our solution enhances the original program
//from TIJ4 with an assembly line elaboration by Doug Lea
//(http://gee.cs.oswego.edu/dl/cpj/assembly.html).
//Lea writes: “Polling intrinsically relies on busy-wait loops, which are
//intrinsically wasteful (but still sometimes less so than context-switching).
//Coping with this requires empirically guided decisions about how to insert
//sleeps, yields, or alternative actions to strike a balance between conserving
//CPU time and maintaining acceptable average response latencies.” We must
//poll each queue in the program with a wait time of 50 milliseconds (inside the
//Merger task), because data could come from any queue, so blocking a specific
//one indefinitely doesn’t work. The timed poll( ) method from the JDK needs no
//busy-wait loop. We add only Alternator and Merger; Toaster2, Butterer2,
//Jammer2 and Eater2 remain the same.